use redis_protocol::resp2::prelude::*;
//use redis_protocol::resp3::decode::streaming::{decode};
//use redis_protocol::resp3::encode::complete::encode_bytes;
use tokio::{net::TcpStream, io::{self, AsyncReadExt, AsyncWrite, AsyncWriteExt}}; 
use bytes::{BytesMut, BufMut, Buf};
use std::io::{Error, ErrorKind};
pub struct Conn<T> {
    stream: T,
    input: BytesMut
}

impl<T:AsyncReadExt + AsyncWrite + std::marker::Unpin> Conn<T> {
    pub async fn read(&mut self) -> io::Result<Frame>
    {
        let mut b:[u8;1024] = [0;1024];
        loop {
            let len = self.stream.read(&mut b[..]).await?;
            if len == 0 {
                return Err(Error::new(ErrorKind::UnexpectedEof, "Client close connection"))
            }
            self.input.put(&b[..len]);

            if self.input.chunk()[0] == 80 {
                self.input.advance(6);
                return Ok(Frame::Array(vec![Frame::SimpleString("ping".to_string())]))
            }
            match decode(&mut self.input) {
                Ok(Some((f, c))) => {
                    //println!("success input{:?}", self.input);
                    self.input.advance(c);
                    return Ok(f)
                },
                Ok(None) => println!("Nothing {}", len), 
                Err(e) => println!("redis err {} input {:?}", e, self.input),
            };
        }
    }
    pub async fn write(&mut self, data: Frame) -> io::Result<()> 
    {
        let mut buf = BytesMut::new();
        encode_bytes(&mut buf, &data).unwrap();
        self.stream.write(&buf).await?;

        Ok(())
    }
}
impl From<TcpStream> for Conn<TcpStream> {
    fn from(s: TcpStream) -> Self {
        Conn { stream: s, input: BytesMut::with_capacity(1024)}
    }
}


#[cfg(test)]
mod test {
    use super::*;
    use tokio::net::TcpListener;
    #[test]
    fn test_create_connection() 
    {
        let server = tokio_test::block_on(TcpListener::bind("0.0.0.0:955")).unwrap();
        let (c, _) = tokio_test::block_on(server.accept()).unwrap();
        let mut cm = Conn::from(c);
        let frame = tokio_test::block_on(cm.read()).unwrap();
        println!("{:?}", frame); 
    }
}